package com.bsj.travel.mqtt.service.impl;

import cn.hutool.core.net.NetUtil;
import com.bsj.travel.cached.redis.RedisCached;
import com.bsj.travel.constant.BrokerConnectConstants;
import com.bsj.travel.def.global.JsonResult;
import com.bsj.travel.def.mqtt.message.SendMqttMsg;
import com.bsj.travel.def.mqtt.message.node.MqttServerNode;
import com.bsj.travel.mqtt.enums.RedisKeys;
import com.bsj.travel.mqtt.service.MqttBrokerService;
import com.bsj.travel.util.StringUtils;
import com.bsj.travel.util.ip.GetEth0IpUtil;
import lombok.extern.slf4j.Slf4j;
import net.dreamlu.iot.mqtt.codec.MqttQoS;
import net.dreamlu.iot.mqtt.core.server.MqttServer;
import net.dreamlu.iot.mqtt.core.server.MqttServerCreator;
import net.dreamlu.mica.core.utils.INetUtil;
import net.dreamlu.mica.redis.cache.MicaRedisCache;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.discovery.DiscoveryClient;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;

import java.util.*;

/**
 * @author yinhao
 * @version 1.0
 * @description mqtt broker服务
 * @date 2023/11/9
 */
@Service
@Slf4j
public class MqttBrokerServiceImpl implements MqttBrokerService {
    @Autowired
    private MicaRedisCache redisCache;

    @Autowired
    private MqttServer mqttServer;

    @Autowired
    private DiscoveryClient discoveryClient;

    @Autowired
    private Environment environment;

    @Autowired
    private RedisCached redisCached;

    @Override
    public List<MqttServerNode> getNodes() {
        Map<Object, Object> map = redisCache.hGetAll(RedisKeys.SERVER_NODES.getKey());
        if (map.isEmpty()) {
            return Collections.emptyList();
        }
        List<MqttServerNode> list = new ArrayList<>();
        map.entrySet().iterator().forEachRemaining(item -> {
            Object value = item.getValue();
            list.add(MqttServerNode.builder()
                    .name(item.getKey().toString())
                    .peerHost(value == null ? "" : value.toString())
                    .build());
        });
        return list;
    }

    @Override
    public long getOnlineClientSize() {
        Map<Object, Object> map = redisCache.hGetAll(RedisKeys.SERVER_NODES.getKey());
        if (map.isEmpty()) {
            return 0L;
        }
        long count = 0;
        Iterator<Map.Entry<Object, Object>> iterator = map.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<Object, Object> entry = iterator.next();
            String key = entry.getKey().toString();
            Long size = redisCache.hLen(RedisKeys.CONNECT_STATUS.getKey(key));
            count += size;
        }
        return count;
    }

    @Override
    public long getOnlineClientSize(String nodeName) {
        return redisCache.hLen(RedisKeys.CONNECT_STATUS.getKey(nodeName));
    }

    /**
     * 用于内部集群间MQTT消息下行
     * 1.先进行节点跟发送的客户端匹配，匹配上直接发送消息，如果匹配不上则到注册中心查询
     *
     * @param sendMqttMsg
     * @return
     */
    @Override
    public Boolean sendMsg(SendMqttMsg sendMqttMsg) {
//        MqttServerCreator serverCreator = mqttServer.getServerCreator();
//        String nodeName = serverCreator.getNodeName();
        String nodeName = GetEth0IpUtil.getHostAddress();
        String clientId = redisCache.hGet(RedisKeys.CONNECT_STATUS.getKey(nodeName), sendMqttMsg.getUsername());
        if (StringUtils.isNotNull(clientId)) {
            //直接在自己节点发送
            boolean publish = mqttServer.publish(clientId, sendMqttMsg.getTopic(),
                    sendMqttMsg.getPayload().getBytes(), MqttQoS.valueOf(sendMqttMsg.getQos()));
            return publish;
        }
        Map<Object, Object> map = redisCache.hGetAll(RedisKeys.SERVER_NODES.getKey());
        if (map.isEmpty()) {
            return false;
        }
        //找到对应的长连接在哪个节点中
        ServiceInstance instance = getServiceInst(map, nodeName, sendMqttMsg.getUsername());
        if (instance != null) {
            String url = instance.getUri() + BrokerConnectConstants.BROKER_SEND_MSG;
            RestTemplate restTemplate = new RestTemplate();
            JsonResult forObject = restTemplate.postForObject(url, sendMqttMsg, JsonResult.class);
            Boolean data = (Boolean) forObject.getData();
            return data;
        }
        return false;
    }

    /**
     * 用于内部Mqtt连接状态管理
     *
     * @param username
     * @return
     */
    @Override
    public Boolean mqttConnectTake(String username) {
        MqttServerCreator serverCreator = mqttServer.getServerCreator();
        String clientStr = redisCache.hGet(RedisKeys.CONNECT_STATUS.getKey(serverCreator.getNodeName()), username);
        if (clientStr != null) {
            //直接在自己节点操作
            mqttServer.close(clientStr);
            return true;
        }
        Map<Object, Object> map = redisCache.hGetAll(RedisKeys.SERVER_NODES.getKey());
        if (map.isEmpty()) {
            return false;
        }
        //找到对应的长连接在哪个节点中
        ServiceInstance instance = getServiceInst(map, serverCreator.getNodeName(), username);
        if (instance != null) {
            String url = instance.getUri() + BrokerConnectConstants.BROKER_MQTT_CONNECT_TAKE;
            SendMqttMsg sendMqttMsg = SendMqttMsg.builder().username(username).build();
            RestTemplate restTemplate = new RestTemplate();
            JsonResult forObject = restTemplate.postForObject(url, sendMqttMsg, JsonResult.class);
            Boolean data = (Boolean) forObject.getData();
            return data;
        }
        return false;
    }

    /**
     * 获取对应节点的ServiceInstance
     *
     * @param map      集群内MQTT服务端节点集合
     * @param nodeName 当前节点nodeName
     * @param username mqtt连接username
     * @return ServiceInstance
     */
    private ServiceInstance getServiceInst(Map<Object, Object> map, String nodeName,
                                           String username) {
        String clientId = null;
        ServiceInstance instance = null;
        Iterator<Map.Entry<Object, Object>> iterator = map.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<Object, Object> entry = iterator.next();
            //"3832841@VM-16-14-centos:1883"
            String key = entry.getKey().toString();
            //"172.16.16.14:1883"
            String value = entry.getValue().toString();
            if (key.equalsIgnoreCase(nodeName)) {
                continue;
            }
            clientId = redisCache.hGet(RedisKeys.CONNECT_STATUS.getKey(key), username);
            if (StringUtils.isNull(clientId)) {
                //不在此节点，略过
                continue;
            }
            //找到了这个客户端连接在此节点中
            String applicationName = environment.getProperty("spring.application.name");
            List<ServiceInstance> instances = discoveryClient.getInstances(applicationName);
            //从nacos注册中心获取到当前实例的的所有服务,匹配对应的节点信息
            Optional<ServiceInstance> firstOne = instances.stream().filter(inst -> value.startsWith(inst.getHost())).findFirst();
            if (firstOne.isEmpty()) {
                continue;
            }
            //筛选出对应的服务信息
            instance = firstOne.get();
        }
        return instance;
    }
}
